package com.sc.nft.admin.consume;

import com.sc.nft.dto.NationalTreasuryDTO;
import com.sc.nft.enums.MessageTopic;
import com.sc.nft.enums.RedisKeyEnum;
import com.sc.nft.exception.GlobalRunTimeException;
import com.sc.nft.service.EcologyPoolSourceService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.concurrent.TimeUnit;

@RocketMQMessageListener(topic = MessageTopic.NATIONAL_TREASURY + "${spring.profiles.active}", consumerGroup = MessageTopic.NATIONAL_TREASURY + "${spring.profiles.active}")
@Component
@Slf4j
@RequiredArgsConstructor
public class NationalTreasuryConsumer implements RocketMQListener<NationalTreasuryDTO> {
    private final Redisson redisson;
    private final EcologyPoolSourceService ecologyPoolSourceService;

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void onMessage(NationalTreasuryDTO message) {
        // 上锁,检查该数据在数据库中是否存在
        // 不存在则插入,存在则丢弃不在操作
        // 解锁
        String key = RedisKeyEnum.POINTS_DISTRIBUTION.getKey()+message.getPkCode();
        RLock lock = redisson.getLock(key);
        try {
            if (!lock.tryLock(3, TimeUnit.MINUTES)) {
                throw new GlobalRunTimeException("消费mq消息失败");
            }
            ecologyPoolSourceService.exchequerPointsMQ(message);

        } catch (Exception e) {
            throw new GlobalRunTimeException(e.getMessage());
        }finally {
            if (lock.isLocked()&&lock.isHeldByCurrentThread()) {
                lock.unlock();
            }
        }
    }
}
